package com.shujia.wyh.base;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import javax.swing.*;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/*
    过滤器的使用
 */
public class HbaseFilterDemo {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        try {
            //获取与Hbase的连接对象
            //1、获取Hadoop的相关环境
            Configuration conf = new Configuration();

            //2、设置以下zookeeper的集群信息
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

            //3、获取habse的连接对象
            conn = ConnectionFactory.createConnection(conf);

            //4、获取HMaster的对象
            admin = conn.getAdmin();

            System.out.println("获取连接成功！！" + conn);
            System.out.println("HMaster的对象获取成功！！" + admin);
            System.out.println("------------------------------------------------------------------");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     *  需求1：通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来
     */
    @Test
    public void rowFilterAndBinaryFun(){
        try {
            //将表名封装成TableName的对象
            TableName students = TableName.valueOf("students");

            //获取表的实例
            Table table = conn.getTable(students);

            //创建Scan对象
            Scan scan = new Scan();
            //创建BinaryComparator比较器
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("1500100010"));

            //创建行键过滤
            //老版本：RowFilter(final CompareOp rowCompareOp,final ByteArrayComparable rowComparator)
//            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS, binaryComparator);

            //新版本：public RowFilter(final CompareOperator op,final ByteArrayComparable rowComparator)
            RowFilter rowFilter = new RowFilter(CompareOperator.LESS, binaryComparator);


            //设置过滤器
            scan.setFilter(rowFilter);

            ResultScanner resultScanner = table.getScanner(scan);
            //输出结果
            printResultScanner(resultScanner);


        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 需求：在hbase中创建一张表,有多个列簇
     */
    @Test
    public void createTable() {
        try {
            //在Hbase中创建一张表必须项是表名和列簇名
            //表在API中对应一个表描述器单独创建，列簇也对应一个列簇描述器，单独创建
//            HTableDescriptor
            TableDescriptorBuilder test2 = TableDescriptorBuilder.newBuilder(TableName.valueOf("students2"));

            //创建一个List集合
            ArrayList<ColumnFamilyDescriptor> list = new ArrayList<>();
//            HColumnDescriptor
            ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of("info");
            ColumnFamilyDescriptor shujia = ColumnFamilyDescriptorBuilder.of("shujia");

            list.add(info);
            list.add(shujia);

            //在表中设置列簇
//            test2.setColumnFamily(info);
            test2.setColumnFamilies(list);

            //使用admin来创建表
            admin.createTable(test2.build());

            //输出提示语句，并输出表名
            /**
             * 注意：所有的数据信息在hbase中都被转成了hbase专有的字节数组类型，无法使用java的方式将其转字符串
             * hbase专门提供一个工具类去解析这样的特别的字节数组
             * Bytes
             */
//            System.out.println(Arrays.toString(test2.build().getTableName().getName()) + " 表创建成功！！");
            System.out.println(Bytes.toString(test2.build().getTableName().getName()) + " 表创建成功！！");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求5：将学生信息表
     * 1500100001,施笑槐,22,女,文科六班
     * <p>
     * 考虑：行键怎么设计？
     * 唯一性
     * 长度不宜过长
     * 散列性
     * <p>
     * 对于此刻，将学号作为rowkey
     */
    @Test
    public void putAllData() {
        try {
            //额外的考虑：如果表中有几条脏数据，我们可以先删除脏数据
            //创建Delete对象
//            Delete d1 = new Delete(Bytes.toBytes("1001"));
            //...
            //先将表名封装成TableName对象
            TableName students = TableName.valueOf("students2");
            //获取到表的对象
            Table table = conn.getTable(students);

//            table.delete(d1);
//            System.out.println("成功删除一行数据");
            //提前创建一个List集合对象，封装多个Put对象，每一个put是一列数据
            ArrayList<Put> puts = new ArrayList<>();

            //读取文件
            BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));

            //一次读取一行
            String line = null;
            while ((line = br.readLine()) != null) {
                //1500100001,施笑槐,22,女,文科六班
                //按照英文逗号切
                String[] infos = line.split(",");
                //将学号作为行键创建一个put对象
                Put put = new Put(Bytes.toBytes(infos[0]));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("姓名"), Bytes.toBytes(infos[1]));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("年龄"), Bytes.toBytes(infos[2]));
                put.addColumn(Bytes.toBytes("shujia"), Bytes.toBytes("性别"), Bytes.toBytes(infos[3]));
                put.addColumn(Bytes.toBytes("shujia"), Bytes.toBytes("班级"), Bytes.toBytes(infos[4]));

                //将封装好的Put对象放到集合中
                puts.add(put);
            }

            table.put(puts);
            System.out.println("学生信息添加完毕！！！");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 通过FamilyFilter与SubstringComparator查询列簇名包含in的所有列簇下面的数据
     */
    @Test
    public void FamilyFilterAndSubstringFun(){
        try {
            TableName students = TableName.valueOf("students2");

            Table table = conn.getTable(students);

            Scan scan = new Scan();

            /**
             *  对于同一个过滤器而言，其中的比较器可以是不一样的
             */

            //创建包含比较器
//            SubstringComparator substringComparator = new SubstringComparator("jia");

            //二进制前缀比较器 BinaryPrefixComparator
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("i"));

            //创建列簇过滤器
            //public FamilyFilter(final CompareOperator op, final ByteArrayComparable familyComparator)
//            FamilyFilter familyFilter = new FamilyFilter(CompareOperator.EQUAL, substringComparator);
            FamilyFilter familyFilter = new FamilyFilter(CompareOperator.EQUAL, binaryPrefixComparator);

            scan.setFilter(familyFilter);

            ResultScanner resultScanner = table.getScanner(scan);

            printResultScanner(resultScanner);

        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 通过QualifierFilter与SubstringComparator查询列名包含 ‘级’ 的列的值
     */
    @Test
    public void qualifierFilterAndSubstringFun(){
        try {
            TableName students = TableName.valueOf("students2");

            Table table = conn.getTable(students);

            Scan scan = new Scan();

            //创建包含比较器
            SubstringComparator substringComparator = new SubstringComparator("别");

            //创建列名过滤器
            //public QualifierFilter(final CompareOperator op, final ByteArrayComparable qualifierComparator)
            QualifierFilter qualifierFilter = new QualifierFilter(CompareOperator.EQUAL, substringComparator);

            scan.setFilter(qualifierFilter);

            ResultScanner scanner = table.getScanner(scan);

            printResultScanner(scanner);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "张" 开头的学生
     */
    @Test
    public void valueFilterAndBinaryPrefixFun(){
        try {
            TableName students = TableName.valueOf("students2");

            Table table = conn.getTable(students);

            Scan scan = new Scan();
            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("文科"));

            //创建列值过滤器 只会查询出符合条件的那一列
            //public ValueFilter(final CompareOperator valueCompareOp, final ByteArrayComparable valueComparator)
            ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, binaryPrefixComparator);

            scan.setFilter(valueFilter);

            ResultScanner scanner = table.getScanner(scan);

            printResultScanner(scanner);
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @After
    public void close() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     *  自定义一个方法输出getScanner的结果值
     */
    public void printResultScanner(ResultScanner resultScanner){
        try {
            //定义一个变量是Result类型的
            Result rs = null;
            //将每一次获取的值赋值给rs,如果rs不为null,说明获取到值
            //rs获取的是一行的值，而一行可以有若干个列
            while ((rs = resultScanner.next()) != null) {
                //获取到行键值
                String id = Bytes.toString(rs.getRow());
                //上面的获取，是在我们已经知道了列名的前提下获取的，如果我们不知道列名，又怎么获取呢？
                //获取一行中所有的列，每一列组成了Cell对象，返回一个集合
                List<Cell> cells = rs.listCells();

                StringBuilder sb = new StringBuilder();
                sb.append("学号：").append(id);

                for (Cell cell : cells) {
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    sb.append(", ").append(qualifier).append("：").append(value);
                }

                //得到一行的拼接结果
                System.out.println(sb);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}
